package com.sojson.util.zookeeper.impl;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.Stat;

import com.sojson.constant.Constant;
import com.sojson.util.ExpUtil;
import com.sojson.util.StringUtil;
import com.sojson.util.properties.PropertiesUtil;
import com.sojson.util.text.Convert;
import com.sojson.util.zookeeper.CustomWatcher;
import com.sojson.util.zookeeper.ZookeeperUtil;
import com.sojson.util.zookeeper.ZookeeperUtilInterface;
import com.sojson.util.zookeeper.constant.ZooConstant;

import lombok.extern.slf4j.Slf4j;

/**
 * Zookeeper工具类实现类
 *
 * @author liu
 * @date 2020-11-15
 */
@Slf4j
public class ZookeeperUtilImpl implements ZookeeperUtilInterface {

    static ZookeeperUtilImpl zookeeperUtilImpl;
    /** 会话超时时间(单位: 毫秒,不设置为5秒) */
    public static final Integer SESSION_TIMEOUT = Convert.toInt(
        PropertiesUtil.getNumberFormat(Constant.PROPERTIES.getProperty("so.zookeeper.zookeeper.session.timeout")),
        1000 * 5);
    /** 等待连接成功的最大超时时间(单位: 调方法的时候设置,不设置为10分钟) */
    public static final Integer CONN_TIMEOUT = Convert.toInt(
        PropertiesUtil.getNumberFormat(Constant.PROPERTIES.getProperty("so.zookeeper.zookeeper.connection.timeout")),
        1000 * 60 * 10);
    /** 核心类 */
    private ZooKeeper zk;
    /** 监视器 */
    private CustomWatcher watcher;
    /** 因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程 */
    private CountDownLatch countDownLatch;
    /** Zookeeper服务器地址 */
    private String url;
    /** Acl */
    private Map<String, List<byte[]>> acls;

    {
        acls = new HashMap<>();
        acls.put(ZooConstant.SCHEME_DIGEST, new ArrayList<>());
        acls.put(ZooConstant.SCHEME_IP, new ArrayList<>());
    }

    public ZookeeperUtilImpl() {}

    public ZookeeperUtilImpl(CustomWatcher watcher, CountDownLatch countDownLatch) {
        this.watcher = watcher;
        this.countDownLatch = countDownLatch;
    }

    @Override
    protected void finalize() throws Throwable {
        close();
    }

    /**
     * 获取Zookeeper服务器地址
     *
     * @return
     */
    public String getUrl() {
        return url;
    }

    /**
     * 设置Zookeeper服务器地址
     *
     * @param url
     */
    public void setUrl(String url) {
        this.url = url;
    }

    /**
     * 获取ZooKeeper
     *
     * @return
     */
    public ZooKeeper getZk() {
        return zk;
    }

    /**
     * 设置ZooKeeper
     *
     * @param zk
     */
    public void setZk(ZooKeeper zk) {
        this.zk = zk;
    }

    /**
     * 获取Watcher
     *
     * @return
     */
    public Watcher getWatcher() {
        return watcher;
    }

    /**
     * 设置Watcher
     *
     * @param watcher
     */
    public void setWatcher(CustomWatcher watcher) {
        this.watcher = watcher;
    }

    /**
     * 获取因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程
     *
     * @return
     */
    public CountDownLatch getCountDownLatch() {
        return countDownLatch;
    }

    /**
     * 设置因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程
     *
     * @param countDownLatch
     */
    public void setCountDownLatch(CountDownLatch countDownLatch) {
        this.countDownLatch = countDownLatch;
    }

    /**
     * 设置因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程
     *
     * @param countDownLatch
     */
    public static CountDownLatch createCountDownLatch() {
        return new CountDownLatch(1);
    }

    /**
     * Zookeeper默认监视类
     *
     * @author liu
     * @date 2020-11-15
     */
    static class CustomDefaultWatcher extends ACustomWatcher {

        /** 因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程 */
        private CountDownLatch countDownLatch;
        int i = 1;

        CustomDefaultWatcher(CountDownLatch countDownLatch) {
            this.countDownLatch = countDownLatch;
        }

        /**
         * 监视器(回调方法)
         */
        @Override
        public void process(WatchedEvent event) {
            // 连接完成的时候让计数减一
            if (event.getState() == KeeperState.SyncConnected) {
                getCountDownLatch().countDown();
                return;
            }
        }

        /**
         * 获取因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程
         *
         * @return
         */
        public CountDownLatch getCountDownLatch() {
            return countDownLatch;
        }

        /**
         * 设置因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程
         *
         * @param countDownLatch
         */
        public void setCountDownLatch(CountDownLatch countDownLatch) {
            this.countDownLatch = countDownLatch;
        }
    }

    /**
     * 因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程
     *
     * @param countDownLatch 因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程
     * @return
     */
    private static CustomDefaultWatcher getCustomDefaultWatcher(CountDownLatch countDownLatch) {
        return new CustomDefaultWatcher(countDownLatch);
    }

    /**
     * 获取ZooKeeper
     *
     * @return
     * @throws Exception
     */
    private ZooKeeper getZooKeeper() throws Exception {
        return getZooKeeper(this.url, this.watcher);
    }

    /**
     * 获取ZooKeeper
     *
     * @param url 连接地址
     * @param watcher 监视器
     * @return
     * @throws Exception
     */
    private ZooKeeper getZooKeeper(String url, CustomWatcher watcher) throws Exception {
        ZooKeeper zk = null;
        zk = new ZooKeeper(url, SESSION_TIMEOUT, watcher);
        waitUntilConnected(zk);
        return zk;
    }

    /**
     * 等待连接
     *
     * @param zk
     * @throws Exception
     */
    private void waitUntilConnected(ZooKeeper zk) throws Exception {
        /**
        * 刚创建完立刻打印会话状态为 CONNECTING
        * 线程阻塞5秒,这5秒期间收到了服务端的Watcher通知 SyncConnected
        * 之后会话状态为 CONNECTED
        */
        // 如果连接没有完成就暂停线程
        if (States.CONNECTING == zk.getState()) {
            // 暂停线程直到超时或者计数为0
            this.countDownLatch.await(CONN_TIMEOUT, ZookeeperUtil.TIME_UNIT);
        }
    }

    /**
     * 给Zookeeper添加帐号密码
     *
     * @return
     * @throws Exception
     */
    @Override
    public ZookeeperUtilInterface addAcl() {
        return addAcl(ZookeeperUtil.getAclDigestByte());
    }

    /**
     * 给Zookeeper添加帐号密码
     *
     * @param auth 帐号:密码
     * @return
     * @throws Exception
     */
    @Override
    public ZookeeperUtilInterface addAcl(String auth) {
        return addAcl(auth.getBytes());
    }

    /**
     * 给Zookeeper添加帐号密码
     *
     * @param auth 帐号:密码
     * @return
     * @throws Exception
     */
    @Override
    public ZookeeperUtilInterface addAcl(byte[] auth) {
        List<byte[]> list = acls.get(ZooConstant.SCHEME_DIGEST);
        list.add(auth);
        return this;
    }

    /**
     * 给Zookeeper添加帐号密码
     *
     * @param username 帐号
     * @param password 密码
     * @return
     * @throws Exception
     */
    @Override
    public ZookeeperUtilInterface addAcl(String username, String password) {
        return addAcl(ZookeeperUtil.getAclDigestByte(username, password));
    }

    /**
     * 同步创建一个持久节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public String create(String path) throws Exception {
        return createByte(path, null);
    }

    /**
     * 同步创建一个持久节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String create(String path, String data) throws Exception {
        return createByte(path, data.getBytes());
    }

    /**
     * 同步创建一个持久节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createByte(String path, byte[] data) throws Exception {
        return zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    /**
     * 同步创建一个持久顺序节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public String createSort(String path) throws Exception {
        return createSortByte(path, null);
    }

    /**
     * 同步创建一个持久顺序节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createSort(String path, String data) throws Exception {
        return createSortByte(path, data.getBytes());
    }

    /**
     * 同步创建一个持久顺序节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createSortByte(String path, byte[] data) throws Exception {
        return zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
    }

    /**
     * 同步创建一个有效时间的持久顺序节点
     *
     * @param path      节点
     * @param timeOut   超时时间(单位: 毫秒)
     * @return
     * @throws Exception
     */
    @Override
    public String createSort(String path, long timeOut) throws Exception {
        return createSortByte(path, null, null, timeOut);
    }

    /**
     * 同步创建一个有效时间的持久顺序节点,同时设置节点信息
     *
     * @param path      节点
     * @param stat      节点信息
     * @param timeOut   超时时间(单位: 毫秒)
     * @return
     * @throws Exception
     */
    @Override
    public String createSort(String path, Stat stat, long timeOut) throws Exception {
        return createSortByte(path, null, stat, timeOut);
    }

    /**
     * 同步创建一个有效时间的持久顺序节点
     *
     * @param path      节点
     * @param data   内容
     * @param timeOut   超时时间(单位: 毫秒)
     * @return
     * @throws Exception
     */
    @Override
    public String createSort(String path, String data, long timeOut) throws Exception {
        return createSort(path, data, null, timeOut);
    }

    /**
     * 同步创建一个有效时间的持久顺序节点
     *
     * @param path      节点
     * @param data   内容
     * @param timeOut   超时时间(单位: 毫秒)
     * @return
     * @throws Exception
     */
    @Override
    public String createSortByte(String path, byte[] data, long timeOut) throws Exception {
        return createSortByte(path, data, null, timeOut);
    }

    /**
     * 同步创建一个有效时间的持久顺序节点,同时注册监听
     *
     * @param path      节点
     * @param data   内容
     * @param stat      节点信息
     * @param timeOut   超时时间(单位: 毫秒)
     * @return
     * @throws Exception
     */
    @Override
    public String createSort(String path, String data, Stat stat, long timeOut) throws Exception {
        return createSortByte(path, data.getBytes(), stat, timeOut);
    }

    /**
     * 同步创建一个有效时间的持久顺序节点,同时注册监听
     *
     * @param path      节点
     * @param data   内容
     * @param stat      节点信息
     * @param timeOut   超时时间(单位: 毫秒)
     * @return
     * @throws Exception
     */
    @Override
    public String createSortByte(String path, byte[] data, Stat stat, long timeOut) throws Exception {
        return zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL_WITH_TTL, stat,
            timeOut);
    }

    /**
     * 同步创建一个临时节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public String createTemporary(String path) throws Exception {
        return createTemporaryByte(path, null);
    }

    /**
     * 同步创建一个临时节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createTemporary(String path, String data) throws Exception {
        return createTemporaryByte(path, data.getBytes());
    }

    /**
     * 同步创建一个临时节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createTemporaryByte(String path, byte[] data) throws Exception {
        return zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    }

    /**
     * 同步创建一个临时顺序节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public String createTemporarySort(String path) throws Exception {
        return createTemporarySortByte(path, null);
    }

    /**
     * 同步创建一个临时顺序节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createTemporarySort(String path, String data) throws Exception {
        return createTemporarySortByte(path, data.getBytes());
    }

    /**
     * 同步创建一个临时顺序节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createTemporarySortByte(String path, byte[] data) throws Exception {
        return zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    }

    /**
     * 删除节点
     *
     * @param path 节点
     * @throws Exception
     */
    @Override
    public void delete(String path) throws Exception {
        zk.delete(path, -1);
    }

    /**
     * 修改节点内容
     *
     * @param path 节点
     * @param data 内容
     * @throws Exception
     */
    @Override
    public void setData(String path, byte[] data) throws Exception {
        zk.setData(path, data, -1);
    }

    /**
     * 修改节点内容
     *
     * @param path 节点
     * @param data 内容
     * @throws Exception
     */
    @Override
    public void setData(String path, String data) throws Exception {
        setData(path, data.getBytes());
    }

    /**
     * 获取节点内容(不注册监听)
     *
     * @param path 节点
     * @param stat 通过exists查出来的对象
     * @return
     * @throws Exception
     */
    @Override
    public byte[] getData(String path, Stat stat) throws Exception {
        // 不执行Watcher
        return zk.getData(path, false, stat);
    }

    /**
     * 获取节点内容(同时注册监听)
     *
     * @param path 节点
     * @param stat 通过exists查出来的对象
     * @return
     * @throws Exception
     */
    @Override
    public byte[] getDataTrue(String path, Stat stat) throws Exception {
        // 执行Watcher
        return zk.getData(path, true, stat);
    }

    /**
     * 获取节点内容(同时注册监听)
     *
     * @param path 节点
     * @param watcher 监视器
     * @param stat 通过exists查出来的对象
     * @return
     * @throws Exception
     */
    @Override
    public byte[] getData(String path, CustomWatcher watcher, Stat stat) throws Exception {
        // 执行Watcher
        return zk.getData(path, watcher, stat);
    }

    /**
     * 获取指定节点的所有子节点(不注册监听)
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public List<String> getChildren(String path) throws Exception {
        // 不执行Watcher
        return zk.getChildren(path, false);
    }

    /**
     * 获取指定节点的所有子节点(同时注册监听)
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public List<String> getChildrenTrue(String path) throws Exception {
        // 执行Watcher
        return zk.getChildren(path, true);
    }

    /**
     * 获取指定节点的所有子节点(同时注册监听)
     *
     * @param path 节点
     * @param watcher 监视器
     * @return
     * @throws Exception
     */
    @Override
    public List<String> getChildren(String path, CustomWatcher watcher) throws Exception {
        // 执行Watcher
        return zk.getChildren(path, watcher);
    }

    /**
     * 判断节点是否存在(不注册监听)
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public Stat exists(String path) throws Exception {
        // 不执行Watcher
        return zk.exists(path, false);
    }

    /**
     * 判断节点是否存在(同时注册监听)
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public Stat existsTrue(String path) throws Exception {
        // 执行Watcher
        return zk.exists(path, true);
    }

    /**
     * 判断节点是否存在(同时注册监听)
     *
     * @param path 节点
     * @param watcher 监视器
     * @return
     * @throws Exception
     */
    @Override
    public Stat exists(String path, CustomWatcher watcher) throws Exception {
        // 执行Watcher
        return zk.exists(path, watcher);
    }

    // public static void main(String[] args) throws Exception {
    //
    // ZookeeperUtilImpl instances = getInstancesSingle().open();
    // // ===============================创建会话======================================
    // //
    //
    // /**
    // * 使用第三个构造器，sessionId 和 sessionPasswd用的是上一个连接
    // */
    // long sessionId = instances.zk.getSessionId();
    // byte[] sessionPasswd = instances.zk.getSessionPasswd();
    // @SuppressWarnings("resource")
    // ZooKeeper zookeeper2 = new ZooKeeper("192.168.100.101:2181,192.168.100.102:2181,192.168.100.103:2181", 5000,
    // new CustomDefaultWatcher(createCountDownLatch()), sessionId, sessionPasswd);
    // System.out.println(zookeeper2.getState());
    // Thread.sleep(2000);
    // System.out.println(zookeeper2.getState());
    //
    // // ===============================创建节点======================================
    // /**
    // * CreateMode
    // * PERSISTENT : 持久节点
    // * PERSISTENT_SEQUENTIAL : 持久顺序节点
    // * EPHEMERAL : 临时节点
    // * EPHEMERAL_SEQUENTIAL : 临时顺序节点
    // *
    // * 无论是同步还是异步接口，ZooKeeper都不支持递归创建，即无法在父节点不存在的情况下创建一个子节点。
    // * 另外，如果一个节点已经存在了，那么创建同名节点的时候，会抛出NodeExistException异常。如果是顺序节点,那么永远不会抛出NodeExistException异常
    // * 临时节点不能有子节点,创建节点时,如果给定的父节点是临时节点,则会抛出NoChildrenForEphemeralsException
    // * 创建节点同时，也可以在节点上设置Watcher 当删除节点或者setData时，将会触发Watcher
    // */
    //
    // System.out.println(instances.zk.getState());
    // /**
    // * 同步创建一个持久节点,ACL为 world:anyone:cdrwa 等同于如下命令：
    // * create /node 123 world:anyone:cdrwa
    // */
    // instances.zk.create("/node", "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    //
    // /**
    // * 同步创建一个持久节点，ACL为 world:anyone:cdrwa 所有人只拥有创建的权限,等同于如下命令：
    // * create /node1 123 world:anyone:c
    // */
    // instances.zk.create("/node1", "123".getBytes(),
    // Collections.singletonList(new ACL(ZooDefs.Perms.CREATE, ZooDefs.Ids.ANYONE_ID_UNSAFE)),
    // CreateMode.PERSISTENT);
    //
    // /**
    // * 异步创建一个持久节点, ACL为 digest:wangsaichao:G2RdrM8e0u0f1vNCj/TI99ebRMw=:cdrwa,等同于如下命令：
    // * create /node3 123 digest:wangsaichao:G2RdrM8e0u0f1vNCj/TI99ebRMw=:cdrwa
    // */
    // instances.zk.create("/node3/a", "123".getBytes(),
    // Collections.singletonList(
    // new ACL(ZooDefs.Perms.ALL, new Id("digest", "wangsaichao:G2RdrM8e0u0f1vNCj/TI99ebRMw="))),
    // CreateMode.PERSISTENT, new AsyncCallback.StringCallback() {
    // /**
    // *
    // *
    // * @param rc 返回状态码
    // * @param path 节点全路径
    // * @param ctx 传给服务端的内容,会在异步回调时传回来
    // * @param name 成功时和path一样,除非创建的是顺序节点
    // */
    // @Override
    // public void processResult(int rc, String path, Object ctx, String name) {
    // System.out.println("rc:" + rc);
    // System.out.println("path:" + path);
    // System.out.println("ctx:" + ctx);
    // System.out.println("name:" + name);
    // }
    // }, "传给服务端的内容,会在异步回调时传回来");
    // /**
    // * 注意这里,线程睡眠20秒,可以接收到watcher
    // */
    // Thread.sleep(20000);
    //
    // /**
    // * 创建一个持久顺序定时节点，如果在10000毫秒内 未修改node，并且没有子节点,那么它将被删掉
    // */
    // instances.zk.create("/node4", "123".getBytes(),
    // Collections.singletonList(
    // new ACL(ZooDefs.Perms.ALL, new Id("digest", "wangsaichao:G2RdrM8e0u0f1vNCj/TI99ebRMw="))),
    // CreateMode.PERSISTENT_SEQUENTIAL_WITH_TTL, new AsyncCallback.Create2Callback() {
    // @Override
    // public void processResult(int rc, String path, Object ctx, String name, Stat stat) {
    // System.out.println("rc:" + rc);
    // System.out.println("path:" + path);
    // System.out.println("ctx:" + ctx);
    // System.out.println("name:" + name);
    // System.out.println("stat:" + stat);
    // }
    // }, "传给服务端的内容,会在异步回调时传回来", 10000);
    // Thread.sleep(20000);
    //
    // // ===============================获取节点数据======================================
    //
    // /**
    // * 每一步操作/node3都要先执行该语句
    // * 因为上一步创建的node3 添加了 digest ACL 所以在获取该节点信息的时候,要先添加授权
    // */
    // instances.zk.addAuthInfo("digest", "wangsaichao:123456".getBytes());
    //
    // /**
    // * 同步调用
    // * path 节点路径
    // * Watcher 监视
    // * Stat 节点统计信息
    // * 添加授权之后同步获取节点信息，返回给定路径的节点的数据和统计信息
    // * 如果调用成功,并且watcher参数不为空,则会在具有给定路径的节点上保留监视，当删除节点 或者 setData时候将会触发监视
    // *
    // */
    // byte[] data = instances.zk.getData("/node3", new CustomDefaultWatcher(createCountDownLatch()), new Stat());
    // System.out.println(new String(data));
    // /**
    // * 注意这里,线程睡眠2秒,因为是创建的临时节点,如果不睡眠,你不能使用命令在控制台看见创建的临时节点
    // */
    // Thread.sleep(2000);
    //
    // /**
    // * 异步调用
    // * path 节点路径
    // * watch true使用创建zookeeper时指定的默认watcher 如果为false则不设置监听
    // * DataCallback 异步通知
    // * ctx 回调上下文
    // */
    // instances.zk.getData("/node3", false, new AsyncCallback.DataCallback() {
    // @Override
    // public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
    // System.out.println("rc:" + rc);
    // System.out.println("path:" + path);
    // System.out.println("ctx:" + ctx);
    // System.out.println("data:" + new String(data));
    // System.out.println("stat:" + stat);
    // }
    // }, "传给服务端的内容,会在异步回调时传回来");
    // Thread.sleep(2000);
    //
    // /**
    // * 异步创建一个 临时的顺序节点，ACL为 ip:127.0.0.1:c 等同于如下命令：
    // * create /node2 123 ip:127.0.0.1:c
    // */
    // instances.zk.create("/node2", "123".getBytes(),
    // Collections.singletonList(new ACL(ZooDefs.Perms.CREATE, new Id("ip", "127.0.0.1"))),
    // CreateMode.EPHEMERAL_SEQUENTIAL, new AsyncCallback.StringCallback() {
    // @Override
    // public void processResult(int rc, String path, Object ctx, String name) {
    // System.out.println("rc:" + rc);
    // System.out.println("path:" + path);
    // System.out.println("ctx:" + ctx);
    // System.out.println("name:" + name);
    // }
    // }, "传给服务端的内容,会在异步回调时传回来");
    // /**
    // * 注意这里,线程睡眠20秒,因为是创建的临时节点,如果不睡眠,你不能使用命令在控制台看见创建的临时节点
    // */
    // Thread.sleep(20000);
    //
    // // ===============================修改节点数据======================================
    //
    // /**
    // * setData节点有version这一参数，给定版本与节点的版本匹配，则设置给定路径的节点的数据（如果给定版本是-1，则它匹配任何节点的版本）。返回节点的统计信息。
    // * 如果不存在具有给定路径的节点，则将抛出NoNodeException
    // * 如果给定版本与节点的版本不匹配，将抛出BadVersionException
    // * 设置的数据最大允许大小为1 MB
    // */
    //
    // /**
    // * 每一步操作/node3都要先执行该语句
    // * 因为上一步创建的node3 添加了 digest ACL 所以在获取该节点信息的时候,要先添加授权
    // */
    // instances.zk.addAuthInfo("digest", "wangsaichao:123456".getBytes());
    //
    // /**
    // * 同步设置数据 -1匹配任何版本
    // */
    // Stat stat = instances.zk.setData("/node3", "嗨喽".getBytes(), -1);
    // System.out.println(stat);
    //
    // /**
    // * 异步设置数据
    // */
    // instances.zk.setData("/node3", "helloword".getBytes(), -1, new AsyncCallback.StatCallback() {
    // @Override
    // public void processResult(int rc, String path, Object ctx, Stat stat) {
    //
    // System.out.println("rc:" + rc);
    // System.out.println("path:" + path);
    // System.out.println("ctx:" + ctx);
    // System.out.println("stat:" + stat);
    //
    // }
    // }, "传给服务端的内容,会在异步回调时传回来");
    // Thread.sleep(2000);
    //
    // // ===============================删除节点======================================
    //
    // /**
    // * 删除给定路径的节点。如果存在这样的节点，则调用将成功，并且给定版本与节点的版本匹配（如果给定版本为-1，则它匹配任何节点的版本）。
    // * 如果节点不存在，将抛出NoNodeException。
    // * 如果给定版本与节点的版本不匹配，将抛出BadVersionException。
    // * 如果节点有子节点，将抛出NotEmptyException。
    // * 如果成功将触发现有API调用留下的给定路径节点上的所有监视，以及getChildren API调用留下的父节点上的监视。
    // */
    //
    // /**
    // * 同步删除节点
    // * path 节点路径
    // * version 版本号 -1 代表匹配所有版本
    // */
    // instances.zk.delete("/node1", -1);
    //
    // /**
    // * 异步删除节点
    // */
    // instances.zk.delete("/node2", -1, new AsyncCallback.VoidCallback() {
    // @Override
    // public void processResult(int rc, String path, Object ctx) {
    // System.out.println("rc:" + rc);
    // System.out.println("path:" + path);
    // System.out.println("ctx:" + ctx);
    // }
    // }, "传给服务端的内容,会在异步回调时传回来");
    // Thread.sleep(2000);
    //
    // // ===============================判断节点是否存在======================================
    //
    // /**
    // * 返回给定路径的节点的stat。如果不存在这样的节点，则返回null。
    // * 如果wathher非空并且调用成功（不会抛出异常），则会在具有给定路径的节点上保留监视。wather将由创建/删除节点或在节点上设置数据的成功时触发。
    // */
    //
    // /**
    // * 同步检查节点是否存在,并留下监听
    // */
    // Stat exists = instances.zk.exists("/node2", new Watcher() {
    // @Override
    // public void process(WatchedEvent event) {
    // System.out.println("留下监视");
    // }
    // });
    // System.out.println("判断节点是否存在：" + exists);
    //
    // /**
    // * 异步检查节点是否存在,并留下监听
    // */
    // instances.zk.exists("/node2", new Watcher() {
    // @Override
    // public void process(WatchedEvent event) {
    // System.out.println("留下监视");
    // }
    // }, new AsyncCallback.StatCallback() {
    // @Override
    // public void processResult(int rc, String path, Object ctx, Stat stat) {
    // System.out.println("rc:" + rc);
    // System.out.println("path:" + path);
    // System.out.println("ctx:" + ctx);
    // System.out.println("判断节点是否存在：" + stat);
    // }
    // }, "传给服务端的内容,会在异步回调时传回来");
    //
    // // ===============================ACL操作======================================
    //
    // /**
    // * ACL操作只给 同步 ，异步 自己看文档
    // */
    //
    // /**
    // * 每一步操作/node3都要先执行该语句
    // * 因为上一步创建的node3 添加了 digest ACL 所以在获取该节点信息的时候,要先添加授权
    // */
    //
    // /**
    // * 设置ACL
    // */
    // // 先注册一个 helloworld 密码为123456的用户
    // instances.zk.addAuthInfo("digest", "helloworld:123456".getBytes());
    // // 因为是 /node3 节点 所以还需要 先添加 /node3的授权
    // instances.zk.addAuthInfo("digest", "wangsaichao:123456".getBytes());
    // Stat auth = instances.zk.setACL("/node3",
    // Collections.singletonList(new ACL(ZooDefs.Perms.ALL, new Id("auth", "helloworld:123456"))), -1);
    // System.out.println(auth);
    //
    // /**
    // * 获取ACL
    // */
    // List<ACL> acl = instances.zk.getACL("/node3", new Stat());
    // System.out.println(acl);
    //
    // }

    /**
     * 打开连接
     *
     * @return
     */
    @Override
    public ZookeeperUtilImpl open() {
        try {
            this.zk = getZooKeeper();

            // 添加权限
            Set<Entry<String, List<byte[]>>> entrySet = acls.entrySet();
            for (Entry<String, List<byte[]>> entry : entrySet) {
                String key = entry.getKey();
                List<byte[]> value = entry.getValue();
                for (byte[] val : value) {
                    zk.addAuthInfo(key, val);
                }
            }
        } catch (Exception e) {
            ExpUtil.throwEx(e);
        }
        return this;
    }

    /**
     * 获取ZooKeeper(单例)
     *
     * @return
     */
    public static ZookeeperUtilImpl getInstancesSingle() {
        if (isBlankObject(zookeeperUtilImpl)) {
            CountDownLatch countDownLatch = createCountDownLatch();
            CustomDefaultWatcher customDefaultWatcher = getCustomDefaultWatcher(countDownLatch);
            zookeeperUtilImpl = createInstances(ZookeeperUtil.URL, customDefaultWatcher, countDownLatch);
        }
        return zookeeperUtilImpl;
    }

    /**
     * 获取ZooKeeper
     *
     * @return
     */
    @Override
    public ZookeeperUtilImpl getInstances() {
        CountDownLatch countDownLatch = createCountDownLatch();
        CustomDefaultWatcher customDefaultWatcher = getCustomDefaultWatcher(countDownLatch);
        return createInstances(ZookeeperUtil.URL, customDefaultWatcher, countDownLatch);
    }

    /**
     * 获取ZooKeeper
     *
     * @param url 连接地址
     * @return
     */
    @Override
    public ZookeeperUtilInterface getInstances(String url) {
        CountDownLatch countDownLatch = createCountDownLatch();
        CustomDefaultWatcher customDefaultWatcher = getCustomDefaultWatcher(countDownLatch);
        return getInstances(url, customDefaultWatcher, countDownLatch);
    }

    /**
     * 获取ZooKeeper
     *
     * @param watcher 监视器
     * @param countDownLatch 因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程
     * @return
     */
    @Override
    public ZookeeperUtilImpl getInstances(CustomWatcher watcher, CountDownLatch countDownLatch) {
        return getInstances(ZookeeperUtil.URL, watcher, countDownLatch);
    }

    /**
     * 获取ZooKeeper
     *
     * @param url 连接地址
     * @param watcher 监视器
     * @param countDownLatch 因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程
     * @return
     */
    @Override
    public ZookeeperUtilImpl getInstances(String url, CustomWatcher watcher, CountDownLatch countDownLatch) {
        return createInstances(url, watcher, countDownLatch);
    }

    /**
     * 获取ZooKeeper
     *
     * @param url 连接地址
     * @param watcher 监视器
     * @param countDownLatch 因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程
     * @return
     */
    private static ZookeeperUtilImpl createInstances(String url, CustomWatcher watcher, CountDownLatch countDownLatch) {
        ZookeeperUtilImpl zookeeperUtilImpl = new ZookeeperUtilImpl();
        zookeeperUtilImpl.setUrl(url);
        zookeeperUtilImpl.setWatcher(watcher);
        zookeeperUtilImpl.setCountDownLatch(countDownLatch);
        return zookeeperUtilImpl;
    }

    /**
     * 关闭连接
     *
     * @return
     * @throws Exception
     */
    @Override
    public void close() throws Exception {
        log.info("正在关闭Zookeeper");
        if (zk != null) {
            try {
                zk.close();
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
            zk = null;
        }
    }

    /**
     * 判断一个对象是否是空(空对象)
     *
     * @param obj
     * @return
     */
    private static boolean isBlankObject(Object obj) {
        return StringUtil.isBlankObject(obj);
    }

}